package com.tico.translatereditor.kafkaThread;

import com.alibaba.fastjson.JSONArray;
import com.tico.translatereditor.domain.TaskLinsten;
import com.tico.translatereditor.domain.TranslaterData;
import com.tico.translatereditor.service.KeyValuePairService;
import com.tico.translatereditor.service.TaskLinstenService;
import com.tico.translatereditor.service.TranslaterDataService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;

public class ListenKafkaThread implements Runnable{
    private KeyValuePairService keyValuePairService;
    private TaskLinstenService taskLinstenService;

    public ListenKafkaThread(KeyValuePairService keyValuePairService, TaskLinstenService taskLinstenService) {
        this.keyValuePairService = keyValuePairService;
        this.taskLinstenService = taskLinstenService;
    }

    @Override
    public void run() {
        while (true) {
//            try {
//                Thread.sleep(5000);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
            if(KafkaThread.consumer!=null){
                try {
                    ConsumerRecords<String, String> records = KafkaThread.consumer.poll(1000); //表示每2秒consumer就有机会去轮询一下订阅状态是否需要变更
                    // 本例不关注消息消费，因此每次只是打印订阅结果！
                    System.out.println("当前监听-----"+KafkaThread.consumer.subscription());
                    if (!KafkaThread.subscribedTopics.isEmpty()) {
                        Iterator<String> iter = KafkaThread.subscribedTopics.iterator();
                        List<String> topics = new ArrayList<>();
                        while (iter.hasNext()) {
                            topics.add(iter.next());
                        }
                        KafkaThread.subscribedTopics.clear();
                        KafkaThread.consumer.subscribe(topics); // 重新订阅topic
                    }
                    for (ConsumerRecord<String, String> record : records) {
                        //Thread.sleep(1000);
                        System.out.println("########################################################################################################");
                        System.out.printf("offset = %d, value = %s", record.offset(), record.value(),record.topic());
                        HashMap<String,Object> map = new HashMap<>();
                        map.put("topic",record.topic());
                        List<TaskLinsten> taskLinstens = taskLinstenService.findByNamedParamList(map);
                        if(taskLinstens!=null){
                            String translaterId = taskLinstens.get(0).getTranslaterId();
                            keyValuePairService.translaterMsgToTopic(translaterId,record.value());
                        }
                        System.out.println("########################################################################################################");
                    }
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
    }

    public static void main(String[] args) {
        String arr1 ="[{ \"id\": \"8089483a3342487f855aa03a472a9c46\", \"appointment_id\": \"f12757aec6864f1a9aae3dcb5c1a3971\", \"email\": \"luoxm@tico.cn\", \"uphone\": \"15210566546\", \"uid\": \"5316\", \"uname\": \"雒小梅\" },{ \"id\": \"8089483a3342487f855aa03a472a9c46\", \"appointment_id\": \"f12757aec6864f1a9aae3dcb5c1a3971\", \"email\": \"luoxm@tico.cn\", \"uphone\": \"15210566546\", \"uid\": \"5316\", \"uname\": \"雒小梅\" }]";
        String arr2 ="[{ \"id\": \"8089483a3342487f855aa03a472a9c46\", \"appointment_id\": \"f12757aec6864f1a9aae3dcb5c1a3971\", \"email\": \"luoxm@tico.cn\", \"uphone\": \"13514405474\", \"uid\": \"5316\", \"uname\": \"刘芳明\" }]";
        JSONArray jsonArray = (JSONArray) JSONArray.parse(arr1);
        JSONArray jsonArray1 = (JSONArray) JSONArray.parse(arr2);
        List<String> list = new ArrayList<>();
        List<String> list1 = new ArrayList<>();
        for(int i=0;i<jsonArray.size();i++){
            list.add(jsonArray.getString(i));
        }
        for(int i=0;i<jsonArray1.size();i++) {
            list1.add(jsonArray1.getString(i));
        }
        for(String str :list){
            System.out.println(str);
        }
        System.out.println("-------------------------------------------------");
        for(String str1 :list1){
            System.out.println(str1);
        }
            for(String str :list){
                for(String str1 :list1){
                    if(!str.equals(str1)){

                    }
                }
            }
    }
}
